package com.duobaoyu.dby.supply.message.listener.hander.channel;

import com.alibaba.fastjson.JSON;
import com.duobaoyu.dby.channel.adapter.enums.biz.message.MessageStatusEnum;
import com.duobaoyu.dby.common.mybatis.util.EnumUtils;
import com.duobaoyu.dby.supply.biz.converter.ChannelMessageConverter;
import com.duobaoyu.dby.supply.biz.producer.constant.SupplyChainMQConstants;
import com.duobaoyu.dby.supply.biz.producer.message.ChannelMessage;
import com.duobaoyu.dby.supply.biz.strategy.AbstractChannelMessageAsyncHandler;
import com.duobaoyu.dby.supply.common.biz.bo.message.ChannelMessageBO;
import com.duobaoyu.dby.supply.core.common.exception.CustomAssert;
import com.duobaoyu.dby.supply.core.dinger.annotation.DingerTalk;
import com.duobaoyu.dby.supply.core.framework.util.FrameworkUtils;
import com.duobaoyu.dby.supply.core.rocketmq.handler.AbstractClusterMessageHandler;
import com.duobaoyu.dby.supply.db.entity.message.ChannelMessageOrder;
import com.duobaoyu.dby.supply.db.repository.message.ChannelMessageRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.Optional;

/**
 * @author liyang
 * @version 1.0.0
 * @ClassName SupplyChainChannelMessageHandler
 * @Description 渠道消息异步处理器
 * @createTime 14:53:00 2023/09/28
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class SupplyChainChannelMessageAsyncHandler extends AbstractClusterMessageHandler<ChannelMessage> {

	private final int CONSUME_MAX_TIMES = 2;
	private final ChannelMessageRepository channelMessageRepository;
	private final ChannelMessageConverter channelMessageConverter;

	@Override
	public String getTopic() {
		return SupplyChainMQConstants.SupplyChainTopic.CHANNEL_MESSAGE_TOPIC;
	}

	@Override
	public String getTag() {
		return SupplyChainMQConstants.SupplyChainTopic.ChannelMessageTag.ORDER_MESSAGE_TAG;
	}

	@Override
	@DingerTalk(title = "渠道消息异步处理Consumer")
	public boolean messageHandler(ChannelMessage channelMessage, MessageExt messageExt) {
		if (messageExt.getReconsumeTimes() > CONSUME_MAX_TIMES) {
			return true;
		}
		if (Objects.isNull(channelMessage)) {
			log.warn("渠道消息异步处理器 消息体为空");
			return true;
		}
		String channelMessageId = Optional.of(channelMessage).map(ChannelMessage::getChannelMessageId).orElse(null);
		if (StringUtils.isEmpty(channelMessageId)) {
			log.warn("渠道消息异步处理器 消息ID为空: message : {}", JSON.toJSONString(channelMessage));
			return true;
		}
		com.duobaoyu.dby.supply.db.entity.message.ChannelMessage message = channelMessageRepository.getById(channelMessageId, ChannelMessageOrder.class);
		CustomAssert.notNull(message, "消息记录不存在");
		if (EnumUtils.eq(MessageStatusEnum.SUCCESS, message.getStatus()) || EnumUtils.eq(MessageStatusEnum.NOT_HANDLED, message.getStatus())) {
			if(log.isDebugEnabled()) {
				log.debug("渠道消息异步处理器 消息状态处理成功: message : {}", JSON.toJSONString(message));
			}
			return true;
		}
		ChannelMessageBO bo = channelMessageConverter.toChannelMessage(message);
		AbstractChannelMessageAsyncHandler handler = FrameworkUtils.strategyInstance(AbstractChannelMessageAsyncHandler.class, bo.getChannelCode(), bo.getMessageType());
		CustomAssert.notNull(handler, String.format("渠道消息异步处理器[AbstractChannelMessageAsyncHandler]不存在(channelCode:%s , messageType:%s)", bo.getChannelCode(), bo.getMessageType()));
		return handler.strategyHandler(bo);
	}


}
